package cn.kgc.gmall.app.dws;

import cn.kgc.gmall.app.func.DimAsyncFunction;
import cn.kgc.gmall.bean.OrderWide;
import cn.kgc.gmall.bean.PaymentInfo;
import cn.kgc.gmall.bean.PaymentWide;
import cn.kgc.gmall.bean.ProductStats;
import cn.kgc.gmall.common.GmallConstant;
import cn.kgc.gmall.utils.ClickhouseUtil;
import cn.kgc.gmall.utils.DateTimeUtil;
import cn.kgc.gmall.utils.MyKafkaUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 商品主题 统计
 */
public class ProductStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 0.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(4);
        /*
        //检查点CK相关设置
        env.enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        StateBackend fsStateBackend = new FsStateBackend(
                "hdfs://hadoop202:8020/gmall/flink/checkpoint/ProductStatsApp");
        env.setStateBackend(fsStateBackend);
        System.setProperty("HADOOP_USER_NAME","atkgc");
        */

        //TODO 1.从Kafka中获取数据流
        String groupId = "product_stats_app";
        // 点击 曝光 收藏 加购 下单 支付 退款 评价
        String pageViewSourceTopic = "dwd_topic_page";  // 点击 曝光
        String orderWideSourceTopic = "dwm_order_wide"; // 收藏  下单
        String paymentWideSourceTopic = "dwm_payment_wide";  // 支付
        String cartInfoSourceTopic = "dwd_cart_info";  // 加购
        String favorInfoSourceTopic = "dwd_favor_info"; // 优惠券
        String refundInfoSourceTopic = "dwd_order_refund_info"; // 退款
        String commentInfoSourceTopic = "dwd_comment_info"; // 评价
        FlinkKafkaConsumer<String> pageViewSource  = MyKafkaUtils.getKafkaConsumerSource(pageViewSourceTopic,groupId);
        FlinkKafkaConsumer<String> orderWideSource  = MyKafkaUtils.getKafkaConsumerSource(orderWideSourceTopic,groupId);
        FlinkKafkaConsumer<String> paymentWideSource  = MyKafkaUtils.getKafkaConsumerSource(paymentWideSourceTopic,groupId);
        FlinkKafkaConsumer<String> favorInfoSourceSouce  = MyKafkaUtils.getKafkaConsumerSource(favorInfoSourceTopic,groupId);
        FlinkKafkaConsumer<String> cartInfoSource  = MyKafkaUtils.getKafkaConsumerSource(cartInfoSourceTopic,groupId);
        FlinkKafkaConsumer<String> refundInfoSource  = MyKafkaUtils.getKafkaConsumerSource(refundInfoSourceTopic,groupId);
        FlinkKafkaConsumer<String> commentInfoSource  = MyKafkaUtils.getKafkaConsumerSource(commentInfoSourceTopic,groupId);

        DataStreamSource<String> pageViewDStream = env.addSource(pageViewSource);
        DataStreamSource<String> favorInfoDStream = env.addSource(favorInfoSourceSouce);
        DataStreamSource<String> orderWideDStream= env.addSource(orderWideSource);
        DataStreamSource<String> paymentWideDStream= env.addSource(paymentWideSource);
        DataStreamSource<String> cartInfoDStream= env.addSource(cartInfoSource);
        DataStreamSource<String> refundInfoDStream= env.addSource(refundInfoSource);
        DataStreamSource<String> commentInfoDStream= env.addSource(commentInfoSource);



        // 通过json转化为 实体类
        // TODO 2. 处理页面数据中的商店点击数   处理页面数据中曝光数
        SingleOutputStreamOperator<ProductStats> pageViewProcess = pageViewDStream.process(new ProcessFunction<String, ProductStats>() {
            @Override
            public void processElement(String value, ProcessFunction<String, ProductStats>.Context ctx, Collector<ProductStats> out) throws Exception {
                // 先将字符串转化为 json对象
                // 处理页面数据中的商店点击数
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject page = jsonObject.getJSONObject("page");
                // 获取页面id
                String page_id = page.getString("page_id");
                // 获取时间
                Long ts = jsonObject.getLong("ts");

                // 如果page_id  是空
                if (page_id == null) {
                    System.out.println(jsonObject);
                }

                // 如果不为空 判断是否等于 good_detail
                if ("good_detail".equals(page_id)) {
                    Long sku_id = page.getLong("item");
                    ProductStats productStats = ProductStats.builder().click_ct(1L).ts(ts).sku_id(sku_id).build();
                    out.collect(productStats);
                }

                // 处理页面数据中曝光数
                JSONArray displays = jsonObject.getJSONArray("displays");
                // 遍历曝光数据
                if (displays != null && displays.size() != 0) {
                    for (int i = 0; i < displays.size(); i++) {
                        JSONObject displayJsonObj = displays.getJSONObject(i);

                        String item_type = displayJsonObj.getString("item_type");

                        // 判断曝光数据是否是商品曝光
                        if ("sku_id".equals(item_type)) {
                            Long sku_id = displayJsonObj.getLong("item");
                            ProductStats productStats = ProductStats.builder().sku_id(sku_id).ts(ts).display_ct(1L).build();
                            out.collect(productStats);
                        }
                    }
                }
            }
        });

        // TODO 3.处理下单信息
        SingleOutputStreamOperator<ProductStats> orderWideMap = orderWideDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String value) throws Exception {
                System.out.println(value);
                // 将json字符串转化为对象
                OrderWide orderWide = JSON.parseObject(value, OrderWide.class);
                String create_time = orderWide.getCreate_time();
                // 转化为ts
                Long ts = DateTimeUtil.toTs(create_time);

                return ProductStats
                        .builder()
                        .sku_id(orderWide.getSku_id())
                        .order_sku_num(orderWide.getSku_num())
                        .order_amount(orderWide.getSplit_total_amount())
                        .orderIdSet(new HashSet(Collections.singleton(orderWide.getOrder_id())))
                        .ts(ts)
                        .build();
            }
        });
        // TODO 4.处理收藏数据
        SingleOutputStreamOperator<ProductStats> favorInfoMap = favorInfoDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String value) throws Exception {
                System.out.println(value);
                JSONObject favorInfo = JSON.parseObject(value);
                Long sku_id = favorInfo.getLong("sku_id");
                String create_time = favorInfo.getString("create_time");
                Long ts = DateTimeUtil.toTs(create_time);
                return ProductStats
                        .builder()
                        .sku_id(sku_id)
                        .favor_ct(1L)
                        .ts(ts)
                        .build();
            }
        });

        // TODO 5.处理购物车数据
        SingleOutputStreamOperator<ProductStats> cartInfoMap = cartInfoDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String value) throws Exception {
                System.out.println(value);
                JSONObject cartInfo = JSON.parseObject(value);
                Long sku_id = cartInfo.getLong("sku_id");
                String create_time = cartInfo.getString("create_time");
                Long ts = DateTimeUtil.toTs(create_time);
                return ProductStats
                        .builder()
                        .sku_id(sku_id)
                        .cart_ct(1L)
                        .ts(ts)
                        .build();
            }
        });
        // TODO 6.处理支付数据
        SingleOutputStreamOperator<ProductStats> paymentWideMap = paymentWideDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String value) throws Exception {
                System.out.println(value);
                PaymentWide paymentWide = JSON.parseObject(value, PaymentWide.class);
                // 支付金额  支付回调时间 支付订单号 skuid
                String callback_time = paymentWide.getCallback_time();
                Long ts = DateTimeUtil.toTs(callback_time);
                return ProductStats
                        .builder()
                        .sku_id(paymentWide.getSku_id())
                        .payment_amount(paymentWide.getTotal_amount())
                        .orderIdSet(new HashSet(Collections.singleton(paymentWide.getOrder_id())))
                        .ts(ts)
                        .build();
            }
        });
        // TODO 7.退款  退款订单号 退款金额 退款时间  sku

        SingleOutputStreamOperator<ProductStats> refundInfoMap = refundInfoDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String value) throws Exception {
                System.out.println(value);
                JSONObject refundInfo = JSON.parseObject(value);
                String create_time = refundInfo.getString("create_time");
                Long ts = DateTimeUtil.toTs(create_time);
                return ProductStats.builder()
                        .sku_id(refundInfo.getLong("sku_id"))
                        .ts(ts)
                        .refund_amount(refundInfo.getBigDecimal("refund_amount"))
                        .orderIdSet(new HashSet(Collections.singleton(refundInfo.getString("order_id"))))
                        .build();
            }
        });

        // TODO 8.评价
        SingleOutputStreamOperator<ProductStats> commentInfoMap = commentInfoDStream.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String value) throws Exception {
                System.out.println(value);
                JSONObject commentInfo = JSON.parseObject(value);
                String appraise = commentInfo.getString("appraise");
                // 如果是好评 好评数加1
                Long good_comment_ct = GmallConstant.APPRAISE_GOOD.equals(appraise) ? 1L : 0L;
                // 时间
                String create_time = commentInfo.getString("create_time");
                Long ts = DateTimeUtil.toTs(create_time);
                return ProductStats.builder()
                        .sku_id(commentInfo.getLong("sku_id"))
                        .good_comment_ct(good_comment_ct)
                        .comment_ct(1L)
                        .ts(ts)
                        .build();
            }
        });

        // TODO 9. 将7条已经转化好的流进行union
        DataStream<ProductStats> productStatsDS = pageViewProcess.union(
                orderWideMap, cartInfoMap, refundInfoMap, commentInfoMap, favorInfoMap, paymentWideMap
        );
        // TODO 10.设置水位线

        SingleOutputStreamOperator<ProductStats> productWatermarkStatsDS = productStatsDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<ProductStats>forMonotonousTimestamps().withTimestampAssigner(
                        new SerializableTimestampAssigner<ProductStats>() {
                            @Override
                            public long extractTimestamp(ProductStats element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }
                )
        );

        //productWatermarkStatsDS.print();

        // TODO 11.分组
        KeyedStream<ProductStats, Long> productStatsKeyBy = productWatermarkStatsDS.keyBy(new KeySelector<ProductStats, Long>() {
            @Override
            public Long getKey(ProductStats value) throws Exception {
                //System.out.println("::::::"+value.getSku_id());
                return value.getSku_id();
            }
        });

        // TODO 12.开窗
       WindowedStream<ProductStats, Long, TimeWindow> productStatsWindow = productStatsKeyBy.window(TumblingEventTimeWindows.of(Time.seconds(10L)));

        // TODO 13.聚合
        SingleOutputStreamOperator<ProductStats> productStatsReduce = productStatsWindow.reduce(
                new ReduceFunction<ProductStats>() {
                    @Override
                    public ProductStats reduce(ProductStats ps1, ProductStats ps2) throws Exception {
                        // 曝光聚合
                        ps1.setDisplay_ct(ps1.getDisplay_ct() + ps2.getDisplay_ct());
                        // 点击聚合
                        ps1.setClick_ct(ps1.getClick_ct() + ps1.getClick_ct());
                        // 收藏聚合
                        ps1.setFavor_ct(ps1.getFavor_ct() + ps1.getFavor_ct());
                        // 加购聚合
                        ps1.setCart_ct(ps1.getCart_ct() + ps1.getCart_ct());
                        // 下单商品个数
                        ps1.setOrder_sku_num(ps1.getOrder_sku_num() + ps2.getOrder_sku_num());
                        // 将所有订单id放入一个集合中 计算订单id
                        ps1.getOrderIdSet().addAll(ps2.getOrderIdSet());
                        ps1.setOrder_ct(ps1.getOrderIdSet().size() + 0L);
                        // 下单商品金额
                        ps1.setOrder_amount(ps1.getOrder_amount().add(ps2.getOrder_amount()));

                        // 支付金额
                        ps1.setPayment_amount(ps1.getPayment_amount().add(ps2.getPayment_amount()));
                        // 将所有支付订单id放入一个集合中
                        ps1.getPaidOrderIdSet().addAll(ps2.getPaidOrderIdSet());
                        // 支付订单数
                        ps1.setPaid_order_ct(ps1.getPaidOrderIdSet().size() + 0L);
                        // 将所有退款订单id放入一个集合中
                        ps1.getRefundOrderIdSet().addAll(ps2.getRefundOrderIdSet());
                        ps1.setRefund_order_ct(ps1.getRefundOrderIdSet().size() + 0L);
                        // 退款金额
                        ps1.setRefund_amount(ps1.getRefund_amount().add(ps2.getRefund_amount()));
                        // 评论数
                        ps1.setComment_ct(ps1.getComment_ct() + ps2.getComment_ct());
                        // 好评
                        ps1.setGood_comment_ct(ps1.getGood_comment_ct() + ps2.getGood_comment_ct());

                        return ps1;
                    }

                },
                // 处理窗口时间
                new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                    @Override
                    public void process(Long aLong, ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>.Context context, Iterable<ProductStats> elements, Collector<ProductStats> out) throws Exception {
                        // 获取窗口时间
                        long start = context.window().getStart();
                        long end = context.window().getEnd();
                        for (ProductStats element : elements) {
                            element.setStt(DateTimeUtil.toYMDhms(new Date(start)));
                            element.setEdt(DateTimeUtil.toYMDhms(new Date(end)));
                            out.collect(element);
                        }
                    }
                }
        );


        //productStatsReduce.print();

        // TODO 14.从hbase中获取其他维度信息
        // 14.1. 补充sku维度信息
        SingleOutputStreamOperator<ProductStats> productStatsAsyncSKU = AsyncDataStream.unorderedWait(
                productStatsReduce,
                new DimAsyncFunction<ProductStats>("DIM_SKU_INFO") {

                    @Override
                    protected void join(ProductStats productStats, JSONObject dimInfo) {
                        // 商品sku名称
                        productStats.setSku_name(dimInfo.getString("SKU_NAME"));
                        // 商品价格
                        productStats.setSku_price(dimInfo.getBigDecimal("PRICE"));
                        // spu名称
                        productStats.setSpu_id(dimInfo.getLong("SPU_ID"));
                        // 品牌编号
                        productStats.setTm_id(dimInfo.getLong("TM_ID"));
                        // 品类编号
                        productStats.setCategory3_id(dimInfo.getLong("CATEGORY3_ID"));

                    }

                    @Override
                    protected String getKey(ProductStats input) {
                        return input.getSku_id().toString();
                    }
                },
                600,
                TimeUnit.SECONDS
        );

        // 14.2 同步spu名称
        SingleOutputStreamOperator<ProductStats> productStatsAsyncSPU = AsyncDataStream.unorderedWait(
                productStatsAsyncSKU,
                new DimAsyncFunction<ProductStats>("DIM_SPU_INFO") {
                    @Override
                    protected void join(ProductStats productStats, JSONObject dimInfo) {
                        String spu_name = dimInfo.getString("SPU_NAME");
                        productStats.setSpu_name(spu_name);
                    }

                    @Override
                    protected String getKey(ProductStats input) {
                        return input.getSpu_id().toString();
                    }
                },
                600,
                TimeUnit.SECONDS
        );

        // 14.3 同步品牌名称
        SingleOutputStreamOperator<ProductStats> productStatsAsyncTm = AsyncDataStream.unorderedWait(
                productStatsAsyncSPU,
                new DimAsyncFunction<ProductStats>("DIM_BASE_TRADEMARK") {
                    @Override
                    protected void join(ProductStats productStats, JSONObject dimInfo) {
                        String tm_name = dimInfo.getString("TM_NAME");
                        productStats.setTm_name(tm_name);
                    }

                    @Override
                    protected String getKey(ProductStats input) {
                        return input.getTm_id().toString();
                    }
                },
                600,
                TimeUnit.SECONDS
        );
        // 14.4 同步品类名称
        SingleOutputStreamOperator<ProductStats> productStatsAsync = AsyncDataStream.unorderedWait(
                productStatsAsyncTm,
                new DimAsyncFunction<ProductStats>("DIM_BASE_CATEGORY3") {
                    @Override
                    protected void join(ProductStats productStats, JSONObject dimInfo) {
                        String name = dimInfo.getString("NAME");
                        productStats.setCategory3_name(name);
                    }

                    @Override
                    protected String getKey(ProductStats input) {
                        return input.getCategory3_id().toString();
                    }
                },
                600,
                TimeUnit.SECONDS
        );
        productStatsAsync.print();
        // 最总数据写出到clickHouse中
        productStatsAsync.addSink(
                ClickhouseUtil.getJdbcSink(
                        "insert into product_stats_2022 values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
                )
        );

        env.execute();
    }
}
